package ex.datastream;

import ex.datastream.functions.richFunction.StatefulKeyedProcessFunc;
import ex.datastream.functions.function.FlatMapFuncBySplitter02;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;


public class Checkpoint02 extends ApiFrame {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        Checkpoint02 point02 = new Checkpoint02();
        point02.getEnv();
        // 全局切断任务链
//        point02.env.disableOperatorChaining();

        // 开启checkpoint,每间隔5秒持久化到磁盘一次，可以实现无限重启
        point02.env.enableCheckpointing(5000);

        // 设置持久化路径
        point02.env.getCheckpointConfig().setCheckpointStorage("file:///tmp/ck");

        //设置故障转移,第一个参数最多重试重启次数，第二个参数两次重启次数的时间间隔
        point02.env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5000));

        //取消任务后仍保留状态
        point02.env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        // 加载数据源
        KafkaSource source = point02.getKafkaSource();
        DataStreamSource<String> dataStreamSource = point02.env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");
        SingleOutputStreamOperator outputStreamOperator1=dataStreamSource.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                System.out.println("filter >>"+s);
                return s.contains("a");
            }
        });
        outputStreamOperator1.print("outputStreamOperator1");
        // 处理kafka数据
        SingleOutputStreamOperator<Tuple2<String, Integer>> items = dataStreamSource.flatMap(new FlatMapFuncBySplitter02());//按key进行分组对value求和

        KeyedStream keyedStream = items.keyBy(value -> value.f0);
        SingleOutputStreamOperator outputStreamOperator2 = keyedStream.process(new StatefulKeyedProcessFunc()).slotSharingGroup("3");

        outputStreamOperator2.print("outputStreamOperator2").slotSharingGroup("4");
        point02.env.execute("save checkpoint job");

    }
}
